package jk.fox.node.client;

import io.netty.handler.timeout.TimeoutException;
import jk.fox.node.node.RpcNode;
import jk.fox.node.rule.RuleFactory;
import jk.fox.common.RpcURIUtils;
import jk.fox.common.exception.runtime.RpcRequestTimeoutException;
import jk.fox.common.exception.runtime.RpcServerNoRegistException;
import jk.fox.config.RpcConfiguration;
import jk.fox.engine.RpcNodeDefinition;
import jk.fox.exchange.client.ClientExchange;
import jk.fox.registry.discovery.DiscoveryInfo;
import jk.fox.registry.discovery.NodeDiscovery;
import org.apache.log4j.Logger;

import java.net.URI;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author cuichao
 * @Description: RpcClient
 * @Date: create in 2020/4/17 22:47
 */
public class RpcClient implements RpcCluster {

    private static Logger logger = Logger.getLogger(RpcClient.class);

    private ConcurrentMap<String, RpcNode> nodes = new ConcurrentHashMap<>(32);

    private NodeDiscovery discovery;

    private RuleFactory ruleFactory;

    private ClientExchange exchange;

    private RpcConfiguration config;

    private Lock lock = new ReentrantLock();

    /**
     * 是否允许request超时重试
     */
    private boolean enableRequestRetry;
    /**
     * 请求最大重试次数
     */
    private int requestMaxRetires;
    /**
     * 请求超时时间
     */
    private long requestTimeout;


    public RpcClient(NodeDiscovery discovery, RuleFactory ruleFactory, ClientExchange exchange,RpcConfiguration config) {
        this.discovery = discovery;
        this.ruleFactory = ruleFactory;
        this.exchange = exchange;
        this.config = config;
        this.enableRequestRetry = config.isEnableRequestRetry();
        this.requestMaxRetires = config.getRequestMaxRetires();
        this.requestTimeout = config.getRequestTimeOutMs();
    }

    @Override
    public RpcNode addNode(RpcNodeDefinition rdf){
        String domain = rdf.getId();
        RpcNode node = nodes.get(domain);
        if(node == null ){
            lock.lock();
            try {
                node = nodes.get(domain);
                if(node == null ){
                    //生成RPC服务逻辑节点并配置负载规则
                    node = new RpcNode(rdf,ruleFactory.getRule());
                    node.enableSameSource(config.isEnableSameSource());
                    logger.info("RpcNode is create: "+node.getDomain());
                    //缓存化node
                    nodes.put(domain,node);
                    //订阅发现模块
                    logger.info("RpcNode is subscribe discovery module: "+node.getDomain());
                    DiscoveryInfo item = discovery.getAndSubscribe(node);
                    //合并远程注册表
                    mergeDefinition(rdf,item.getRemoteDefinition());
                    List<String> address = item.getNodes();
                    if(address != null){
                        node.addNode(address);
                    }
                    //监听exchange 连接状态
                    exchange.addServerStatusListener(node);
                    logger.info("RpcNode is subscribe serverStatusListener: "+node.getDomain());
                }
            }finally {
                lock.unlock();
            }
        }
        return node;
    }

    @Deprecated
    public RpcNode addNode(String domain){
        RpcNode node = nodes.get(domain);
        if(node == null ){
            lock.lock();
            try {
                node = nodes.get(domain);
                if(node == null ){
                    //生成RPC服务逻辑节点并配置负载规则
                    node = new RpcNode(domain,ruleFactory.getRule());
                    logger.info("RpcNode is create: "+node.getDomain());
                    //缓存化node
                    nodes.put(domain,node);
                    //订阅发现模块
                    logger.info("RpcNode is subscribe discovery module: "+node.getDomain());
                    DiscoveryInfo item = discovery.getAndSubscribe(node);
                    List<String> address = item.getNodes();
                    if(address != null){
                        node.addNode(address);
                    }
                    //监听exchange 连接状态
                    exchange.addServerStatusListener(node);
                    logger.info("RpcNode is subscribe serverStatusListener: "+node.getDomain());
                }
            }finally {
                lock.unlock();
            }
        }
        return node;
    }

    @Override
    public Object execute(String path,RpcNode node, Object[] parameter,String[] parameterTypes) throws Exception {
        String domain = path.split("/")[0];
        if(node == null ){
            logger.info("RpcNode is not use at node: "+domain);
            throw new RpcServerNoRegistException(String.format("id 为 %s has not register。"));
        }
        //根据rule获取节点地址
        String address = choose(node);
        //构建URI
        URI uri = RpcURIUtils.getUri(address, path);
        Object result = doExecute(node.getDefinition(),uri,parameter,parameterTypes);
        return result;
    }

    @Override
    public String choose(RpcNode node){
        return node.choose();
    }

    @Override
    public Map<String, RpcNode> nodes() {
        return Collections.unmodifiableMap(nodes);
    }

    @Override
    public boolean hasSameSource(String domain){
        RpcNode node = nodes.get(domain);
        return node.hasSameSource();
    }



    private void mergeDefinition(RpcNodeDefinition reference, RpcNodeDefinition service){
        if(service == null){
            return;
        }
        if(reference.getEnableRetry() ==null && service.getEnableRetry() != null){
            reference.setEnableRetry(service.getEnableRetry());
        }
        if(reference.getMaxRetires() == null && service.getMaxRetires() !=null){
            reference.setMaxRetires(service.getMaxRetires());
        }
        if(reference.getTimeoutMS() == null && service.getTimeoutMS() !=null){
            reference.setTimeoutMS(service.getTimeoutMS());
        }
    }


    private Object doExecute(RpcNodeDefinition rnd, URI uri, Object[] parameter, String[] parameterTypes) throws Exception{
        Object result = null;
        boolean enableRetires = isEnableRequestRetry(rnd);
        int maxRetires = requestMaxRetires(rnd);
        int counter = 0;
        long timeout = requestTimeout(rnd);
        do{
            try {
                result = exchange.execute(uri,parameter,parameterTypes,timeout);
            }catch (RpcRequestTimeoutException | TimeoutException  e){
                if( !enableRetires || counter > maxRetires){
                    String exceptionMsg = String.format("%s request is time out", uri);
                    logger.error(exceptionMsg,e);
                    throw new RpcRequestTimeoutException(exceptionMsg);
                    /*break;*/
                }
                logger.debug(String.format("%s request is time out prepare retry it the counter is ： %d", uri,counter+1));
                continue;
            }catch (Exception e){
                throw  e;
            }
        }while (enableRetires && counter++ < maxRetires);

        return result;
    }

    /**
     * 获取request是否能重试
     * @param rnd
     * @return
     */
    protected boolean isEnableRequestRetry(RpcNodeDefinition rnd){
        if(rnd.getEnableRetry() != null){
            //config from RpcNodeDefinition
            return rnd.getEnableRetry();
        }
        //config from Configuration
        return enableRequestRetry;
    }

    /**
     * 获取request 重试次数
     * @param rnd
     * @return
     */
    protected int requestMaxRetires(RpcNodeDefinition rnd){
        if(rnd.getEnableRetry() != null){
            //config from RpcNodeDefinition
            return rnd.getMaxRetires();
        }
        //config from Configuration
        return requestMaxRetires;
    }

    /**
     * 获得request-timeout
     * @param rnd
     * @return
     */
    protected long requestTimeout(RpcNodeDefinition rnd){
        return rnd.getTimeoutMS() == null || rnd.getTimeoutMS() == 0 ? requestTimeout : rnd.getTimeoutMS() ;
    }








}
